package com.zzg.elasticsearch.config;



import java.io.IOException;
import java.net.InetAddress;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

@Configuration
@Component
public class ESConfig {
	private Logger logger = LoggerFactory.getLogger(ESConfig.class);
	
	//es 连接属性
	@Value("${elasticsearch.ip}")
	private String ip;
	@Value("${elasticsearch.port}")
	private String port;
	@Value("${elasticsearch.cluster-name}")
	private String clusterName;
	private static final String poolSize = "20";
	
	private TransportClient  transportClient;
	
	public TransportClient init(){
		logger.info("初始化开始。。。。。");
		if(transportClient == null){
			try {
	            // 配置信息
	            Settings esSetting = Settings.builder()
	                    .put("client.transport.sniff", true)//增加嗅探机制，找到ES集群
	                    .put("thread_pool.search.size", Integer.parseInt(poolSize))//增加线程池个数，暂时设为20
	                    .put("cluster.name",clusterName)
	                    .build();
	            //配置信息Settings自定义,下面设置为EMPTY
	            transportClient = new PreBuiltTransportClient(esSetting);
	            TransportAddress transportAddress = new InetSocketTransportAddress(InetAddress.getByName(ip), Integer.valueOf(port));
	            transportClient.addTransportAddresses(transportAddress);
	 
	        } catch (Exception e) {
	        	e.printStackTrace();
	        	logger.error("elasticsearch TransportClient create error!!!", e);
	        }
		}
		
		return transportClient;
	}
	
	/**
	 * elasticsearch 创建索引
	 */
	
	public IndexResponse  createIndex(String index, String type, String id){
		IndexResponse response = null;
		if(transportClient == null){
			logger.error("elasticsearch TransportClient is null!!!");
			return response;
		}
		IndexRequestBuilder  builder = transportClient.prepareIndex(index, type, id);
		XContentBuilder content = null;
		try {
			// 固定文档格式
			content = XContentFactory.jsonBuilder().startObject()
				.field("user", "zhouzhigang")
				.field("date", new Date())
				.field("content", "elasticsearch")
				.endObject();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			logger.error("elasticsearch XContentBuilder create error!!!", e);
		}
		if(content != null){
			response = builder.setSource(content).get();
		}
		return response;
		
	}
	
	/**
	 * elasticsearch 添加json 文档
	 */
	
	public RestStatus addDocument(String index, String type){
		RestStatus status = null;
		// 文档json 格式
		String json = "{" +
                "\"user\":\"zhouzhiwengang\"," +
                "\"date\":\"2019-01-29\"," +
                "\"content\":\"add Elasticsearch document\"" +
                "}";
		IndexRequestBuilder builder = transportClient.prepareIndex(index, type);
		if(builder ==null){
			logger.error("elasticsearch TransportClient is null!!!");
			return status;
		}
		status = builder.setSource(json, XContentType.JSON).get().status();
		return status;
	}
	
	/**
	 * elasticsearch 根据索引检索
	 */
	public Map findById(String index, String type, String id){
		Map<String,Object> map = null;
		GetResponse builder = transportClient.prepareGet(index, type, id).get();
		if(builder != null){
			map = builder.getSource();
		}
		return map;
	}
	
	/**
	 * elasticsearch 根据索引删除
	 */
	public RestStatus deleteById(String index, String type, String id){
		RestStatus status = null;
		DeleteResponse response = transportClient.prepareDelete(index, type, id).get();
		if(response == null){
			logger.error("elasticsearch DeleteResponse is null!!!");
			return status;
		}
		status = response.status();
		return status;
	}
	
	/**
	 * elasticsearch 根据条件删除
	 */
	public long deleteByQuery(String filed, String value, String index){
		long num = 0L;
		TermQueryBuilder query = new TermQueryBuilder(filed,value);
		BulkByScrollResponse response = DeleteByQueryAction.INSTANCE.newRequestBuilder(transportClient)
			.filter(query)
			.source(index)
			.get();
		if(response == null){
			logger.error("elasticsearch BulkByScrollResponse is null!!!");
			return num;
		}
		num = response.getDeleted();
		return num;
	}
	
	/**
	 * elasticsearch 更新文档指定属性值：根据指定索引ID
	 */
	public RestStatus updateDocument(String index, String type, String id,String field, String value){
		RestStatus status = null;
		UpdateRequest  request = new UpdateRequest ();
		request.index(index);
		request.type(type);
		request.id(id);
		
		XContentBuilder content = null;
		try {
			// 固定文档格式
			content = XContentFactory.jsonBuilder().startObject()
				.field(field, value)
				.endObject();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			logger.error("elasticsearch XContentBuilder create error!!!", e);
			return status;
		}
		
		if(content == null){
			logger.error("elasticsearch XContentBuilder is null!!!");
			return status;
		}
		// UpdateRequest 设置更新文档内容对象
		request.doc(content);
		
		try {
			status = transportClient.update(request).get().status();
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			logger.error("elasticsearch error message!!!", e);
			return status;
		}
		return status;
	}
	
	/**
	 * elasticsearch 批量更新文档对象
	 */
	public void batchUpdate(List<JSONObject> list){
		BulkRequestBuilder batchBuilder = transportClient.prepareBulk();
		if(list != null && list.size() > 0){
			list.stream().forEach(item ->{
				// json 构建索引
				IndexRequestBuilder builder = transportClient.prepareIndex(item.getString("index"), item.getString("type"), item.getString("id"));
				Map<String,String> map = (Map<String, String>) item.get("content");
				
				XContentBuilder content = null;
				try {
					// 固定文档格式
					content = XContentFactory.jsonBuilder().startObject();
					Iterator iter = map.entrySet().iterator();
					while (iter.hasNext()) {
						Map.Entry entry = (Map.Entry) iter.next();
						String key = (String) entry.getKey();
						String value = (String) entry.getValue();
						content.field(key, value);
					}
					content.endObject();
					builder.setSource(content);
					// 批量构建
					batchBuilder.add(builder);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					logger.error("elasticsearch XContentBuilder create error!!!", e);
				}
			});
			BulkResponse response = batchBuilder.get();
			if (response.hasFailures()) {
				BulkItemResponse[] responses = response.getItems();
				for (int i = 0; i < responses.length; i++) {
			          BulkItemResponse item = responses[i];
			            if (item.isFailed()) {
			            	UpdateResponse update = item.getResponse();
			            	GetResult result = update.getGetResult();
			            	// 错误文档-> 转换string
			            	String error = result.sourceAsString();
			            }
			            	
			            }
			        }
			}
	}
	
	/**
	 * elasticsearch 多条件查询
	 */
	
	public void indexTermQueryAndFilter(String index, String type, String fieldName, String value, String rangFieldName, Object from, Object to, int size){
		SearchRequestBuilder  search = transportClient.prepareSearch(index).setTypes(type).setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
		// 属性查询
		TermQueryBuilder termQuery = new TermQueryBuilder(fieldName,value);
		// 范围查询
		RangeQueryBuilder rangQuery = new RangeQueryBuilder(rangFieldName);
		rangQuery.from(from).to(to);
		
		SearchResponse response = search.setQuery(termQuery).setPostFilter(rangQuery).setFrom(0).setSize(size).setExplain(true).get();
		// 输出search 对象
		for (SearchHit hit : response.getHits()) {
			String json = hit.getSourceAsString();
			Object obj = JSONObject.parse(json);
		}
	}
	
	
	
	
	
	
	
}
